package com.bff.gaia.mix.api.xjoin2.client.mysql.async;

import io.vertx.core.Vertx;
import io.vertx.core.VertxOptions;
import io.vertx.core.json.JsonObject;
import io.vertx.ext.jdbc.JDBCClient;
import com.bff.gaia.configuration.Configuration;
import com.bff.gaia.mix.api.xjoin2.client.rdb.async.RdbRichAsyncXjoinFunction;
import com.bff.gaia.mix.api.xjoin2.client.rdb.table.RdbSideTableInfo;
import com.bff.gaia.mix.api.xjoin2.core.side.SideTableInfo;
import com.bff.gaia.mix.api.xjoin2.core.table.FieldInfo;
import com.bff.gaia.mix.api.xjoin2.core.table.JoinInfo;
import com.bff.gaia.mix.api.xjoin2.core.table.StreamTableInfo;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.List;

public class MysqlRdbRichAsyncXjoinFunction extends RdbRichAsyncXjoinFunction {

	private static final long serialVersionUID = -5049135104656817715L;

	private static final Logger LOG = LoggerFactory.getLogger(MysqlRdbRichAsyncXjoinFunction.class);

	private final static String MYSQL_DRIVER = "com.mysql.jdbc.Driver";

	private static final String MYSQL_DRIVER_6 = "com.mysql.cj.jdbc.Driver";

	public MysqlRdbRichAsyncXjoinFunction(StreamTableInfo streamTableInfo, SideTableInfo sideTableInfo, JoinInfo joinInfo, List<FieldInfo> outFieldInfoList) {
		super(streamTableInfo, sideTableInfo, joinInfo, outFieldInfoList);
	}

	@Override
	public void open(Configuration parameters) throws Exception {
		super.open(parameters);
		JsonObject mysqlClientConfig = new JsonObject();
		RdbSideTableInfo rdbSideTableInfo = (RdbSideTableInfo) xjoinInfo.getSideTableInfo();
		mysqlClientConfig.put("url", rdbSideTableInfo.getUrl())
			.put("driver_class", MYSQL_DRIVER_6)
			.put("max_pool_size", DEFAULT_MAX_DB_CONN_POOL_SIZE)
			.put("user", rdbSideTableInfo.getUserName())
			.put("password", rdbSideTableInfo.getPassword());

		VertxOptions vo = new VertxOptions();
		vo.setEventLoopPoolSize(DEFAULT_VERTX_EVENT_LOOP_POOL_SIZE);
		vo.setWorkerPoolSize(DEFAULT_VERTX_WORKER_POOL_SIZE);
		Vertx vertx = Vertx.vertx(vo);
		setRdbSQLClient(JDBCClient.createNonShared(vertx, mysqlClientConfig));
	}
}